package QC.MessageBox.service;

import QC.MessageBox.annotation.APPTable;
import QC.MessageBox.beans.*;
import QC.MessageBox.beans.dto.PageInfo;
import QC.MessageBox.beans.dto.RecentContactFans;
import QC.MessageBox.common.CodeUtil;
import QC.MessageBox.common.Constants;
import QC.MessageBox.common.EmojiUtil;
import QC.MessageBox.common.PeriodicTime;
import QC.MessageBox.enums.ActiveLimit;
import QC.MessageBox.enums.MsgType;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.json.Json;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.asyncsql.AsyncSQLClient;
import io.vertx.ext.sql.ResultSet;
import io.vertx.ext.sql.SQLConnection;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.*;
import java.util.Objects;

/**
 * Created by Andaren on 2017/4/6.
 * Used for: MySQL数据访问
 *      （注意：1.将future.fail(as.cause())抛出在上层统一处理
 *              2.sql操作完成后关闭连接conn.close()，否则几次获取后再次获取getConnection失败
 *      ）
 */
public class MySQLService {

    private static final Logger LOGGER = LoggerFactory.getLogger(MySQLService.class);

    private AsyncSQLClient mySQLClient;

    public MySQLService (AsyncSQLClient mySQLClient) {
        this.mySQLClient = mySQLClient;
    }

    /**
     * 获取连接
     * @return
     */
    public Future<SQLConnection> getConnection() {
        Future<SQLConnection> connFut = Future.future();
        mySQLClient.getConnection(ConnAr -> {
            if (ConnAr.failed()) {
                connFut.fail(ConnAr.cause());
            } else {
                connFut.complete(ConnAr.result());
            }
        });
        return connFut;
    }

    /**
     * 异步关闭连接
     * @param conn
     * @return
     */
    public Future<Void> closeConnection(SQLConnection conn) {
        Future<Void> connFut = Future.future();
        conn.close(closeAr -> {
            if (closeAr.failed()) {
                connFut.fail(closeAr.cause());
            } else {
                connFut.complete(closeAr.result());
            }
        });
        return connFut;
    }

    /**
     * 获取公众号信息
     * @param developerId
     * @return
     */
    public Future<WXPublic> getPublicInfoByDeveloperId(String developerId) {
        Future<WXPublic> retFut = Future.future();
        StringBuilder getPublicSb = new StringBuilder();
        JsonArray params = new JsonArray();

        getPublicSb.append(" SELECT ID, SECRET, ACCESS_TOKEN, APPID, NAME, CHANNEL_TYPE, UPDATETIME, CREATETIME FROM T_PUBLIC ")
                .append(" WHERE DEVELOPERID =  ? ");
        params.add(developerId);
        getPublicSb.append(" ; ");
        query(getPublicSb, params)
                .compose(rows -> {
                    if (rows != null && rows.size() > 0) {
                        WXPublic wxPublic = new WXPublic();
                        wxPublic.setDeveloperId(developerId);
                        for(JsonObject row : rows) {
                            wxPublic.setId(row.getLong("ID"))
                                    .setSecret(row.getString("SECRET"))
                                    .setAccessToken(row.getString("ACCESS_TOKEN"))
                                    .setAppId(row.getString("APPID"))
                                    .setName(row.getString("NAME"))
                                    .setChannelType(row.getInteger("CHANNEL_TYPE"))
                                    .setUpdateTime(row.getString("UPDATETIME"))
                                    .setCreateTime(row.getString("CREATETIME"));
                            retFut.complete(wxPublic);
                            break;
                        }
                    } else {
                        retFut.fail("不存在该公众号developerId=[" + developerId + "]");
                    }
                }, retFut);
        return retFut;
    }

    /**
     * 获取公众号下所有用户的openid列表
     * @param developerId
     * @return
     */
    public Future<List<String>> getAllWxUserByDeveloperId(String developerId) {
        return Future.future(listFut -> {
            StringBuilder uriSb = new StringBuilder();
            JsonArray params = new JsonArray();
            uriSb.append(" SELECT OPENID FROM T_USER_LIMIT WHERE DEVELOPERID = ? ");
            params.add(developerId);
            query(uriSb, params)
                    .compose(rows -> {
                        List<String> openIdList = new ArrayList<>();
                        for (JsonObject row : rows) {
                            openIdList.add(row.getString("OPENID"));
                        }
                        listFut.complete(openIdList);
                    }, listFut);
        });
    }

    /**
     * 获取所有微信公众号信息
     * @return
     */
    public Future<List<WXPublic>> getAllWxPublic() {
        Future<List<WXPublic>> allPublicFuture = Future.future();
        StringBuilder getAllPubSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getAllPubSb.append(" SELECT ID, APPID, SECRET, ACCESS_TOKEN, DEVELOPERID, NAME, CHANNEL_TYPE FROM T_PUBLIC; ");

        query(getAllPubSb, params)
                .compose(rows -> {
                    List<WXPublic> publicList = new ArrayList<>();
                    if (rows != null && rows.size() > 0) {
                        for(JsonObject row : rows) {
                            WXPublic wxPublic = new WXPublic();
                            wxPublic.setId(row.getLong("ID"));
                            wxPublic.setName(row.getString("NAME"));

                            wxPublic.setAppId(row.getString("APPID"));
                            wxPublic.setSecret(row.getString("SECRET"));
                            wxPublic.setAccessToken(row.getString("ACCESS_TOKEN"));
                            wxPublic.setDeveloperId(row.getString("DEVELOPERID"));
                            wxPublic.setChannelType(row.getInteger("CHANNEL_TYPE"));
                            publicList.add(wxPublic);
                        }
                    }
                    allPublicFuture.complete(publicList);
                },allPublicFuture);
        return allPublicFuture;
    }

    /**
     * 更新公众号接口调用token
     * @param developerId   //  公众号原始ID
     * @param access_token  //  新的token
     * @return
     */
    public Future<Void> updateAccessToken(String developerId, String access_token) {
        Future<Void> updateFut = Future.future();
        StringBuilder updateTokenSb = new StringBuilder();
        JsonArray params = new JsonArray();
        updateTokenSb.append(" UPDATE T_PUBLIC SET ACCESS_TOKEN = ?, UPDATETIME = ? WHERE DEVELOPERID = ?; ");
        params
                .add(access_token)
                .add(PeriodicTime.getCurrentTimeStr("yyyy-MM-dd HH:mm:ss"))
                .add(developerId);

        update(updateTokenSb, params)
                .setHandler(updateFut.completer());
        return updateFut;
    }

    /**
     * 获取app端接收消息的用户信息(appUser)
     * @param openId
     * @return
     */
    public Future<AppUser> getCSInfo(String openId) {
        Future<AppUser> csInfofut = Future.future();
        StringBuilder getCsInfoSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getCsInfoSb.append(" SELECT TAU.NICKNAME, TAU.TALKID, TAU.CUSTOMER_SERVER_ID, TAU.WX_OPENID, TAU.CREATETIME, TAU.PULL_BLACK FROM T_APP_USER TAU WHERE TAU.WX_OPENID = ?;  ");
        params.add(openId);

        query(getCsInfoSb, params)
                .compose(rows -> {
                    if (rows.size() == 0) {
                        csInfofut.fail("不存在该用户[" + openId + "]" + "对应的app用户信息");
                    } else {
                        for(JsonObject row : rows) {
                            AppUser newUser = new AppUser();
                            newUser.setCreateTime(row.getString("CREATETIME"))
                                    .setNickname(row.getString("NICKNAME"))
                                    .setTalkId(row.getString("TALKID"))
                                    .setCsId(row.getString("CUSTOMER_SERVER_ID"))
                                    .setOpenId(row.getString("WX_OPENID"))
                                    .setPullBlack(row.getInteger("PULL_BLACK"));
                            csInfofut.complete(newUser);
                            break;
                        }
                    }
                }, csInfofut);
        return csInfofut;
    }

    /**
     * 获取app用户对应openid
     * @param csId
     * @return
     */
    public Future<List<String>> getAppUserOpenId(String csId) {
        Future<List<String>> userOpenidFut = Future.future();
        StringBuilder getOpenIdListSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getOpenIdListSb.append(" SELECT WX_OPENID FROM T_APP_USER WHERE CUSTOMER_SERVER_ID = ?; ");
        params.add(csId);

        query(getOpenIdListSb, params)
                .compose(rows -> {
                    List<String> appUserOpenids = new ArrayList<>();
                    if (rows != null && rows.size() > 0) {
                        for (JsonObject row : rows) {
                            appUserOpenids.add(row.getString("WX_OPENID"));
                        }
                    }
                    userOpenidFut.complete(appUserOpenids);
                }, userOpenidFut);
        return userOpenidFut;
    }

    /**
     * 删除微信用户会话映射
     * @param openId
     * @return
     */
    public Future<Void> deleteAppUserByOpenId(String openId) {
        Future<Void> delAppUserFut = Future.future();
        StringBuilder deleteAppUserSb = new StringBuilder();
        JsonArray params = new JsonArray();
        deleteAppUserSb.append(" DELETE FROM T_APP_USER WHERE WX_OPENID = ?; ");
        params.add(openId);

        update(deleteAppUserSb, params).setHandler(delAppUserFut.completer());
        return delAppUserFut;
    }

    /**
     * 删除微信用户信息
     * @param openId
     * @return
     */
    public Future<Void> deleteWXUserByOpenId(String openId) {
        Future<Void> delWXUserFut = Future.future();
        StringBuilder deleteWXUserSb = new StringBuilder();
        JsonArray params = new JsonArray();
        deleteWXUserSb.append(" DELETE FROM T_WX_USER WHERE WX_OPENID = ?; ");
        params.add(openId);

        update(deleteWXUserSb, params)
                .setHandler(delWXUserFut.completer());
        return delWXUserFut;
    }


    /**
     * 更新 客服id下的所有客户的对应客服关系
     * @param csId
     * @param newCs
     * @return
     */
    public Future<Void> updateAppUserInfoByCsId(String csId, AppUser newCs) {
        Future<Void> updateAppUserTalkIdFut = Future.future();
        StringBuilder updateAppUserTalkIdSb = new StringBuilder();
        JsonArray params = new JsonArray();
        updateAppUserTalkIdSb
                .append(" UPDATE T_APP_USER TAP SET TAP.TALKID = ?, TAP.CUSTOMER_SERVER_ID = ?, TAP.NICKNAME = ? WHERE TAP.CUSTOMER_SERVER_ID = ?; ");
        params
                .add(newCs.getTalkId())
                .add(newCs.getCsId())
                .add(newCs.getNickname())
                .add(csId);

        update(updateAppUserTalkIdSb, params).setHandler(updateAppUserTalkIdFut.completer());
        return updateAppUserTalkIdFut;
    }

    /**
     * 更新 客服id列表下的所有客户的对应客服关系
     * @param openIds
     * @param newCs
     * @return
     */
    public Future<Void> updateAppUserTalkIdByList(JsonArray openIds, AppUser newCs) {
        Future<Void> updateTalkIdInListFut = Future.future();
        StringBuilder updateAppUserTalkIdSb = new StringBuilder();
        JsonArray params = new JsonArray()
                .add(newCs.getTalkId())
                .add(newCs.getCsId())
                .add(newCs.getNickname());
        updateAppUserTalkIdSb
                .append("UPDATE T_APP_USER TAP SET TAP.TALKID = ?, TAP.CUSTOMER_SERVER_ID = ?, TAP.NICKNAME = ? ")
                .append(" WHERE TAP.WX_OPENID IN ( ");
        for (int i = 0; i < openIds.size(); i++) {
            updateAppUserTalkIdSb.append("?").append(",");
            params.add(openIds.getString(i));
        }
        updateAppUserTalkIdSb.deleteCharAt(updateAppUserTalkIdSb.length() - 1);
        updateAppUserTalkIdSb.append(" ); ");

        update(updateAppUserTalkIdSb, params)
                .setHandler(updateTalkIdInListFut.completer());
        return updateTalkIdInListFut;
    }

    /**
     * 获取随机app客服信息
     * @return
     */
    public Future<CustomerServ> getRandomCS() {
        Future<CustomerServ> randCSFut = Future.future();
        mySQLClient.getConnection(as -> {
            if (as.succeeded()) {
                SQLConnection conn = as.result();
                String queryStr = "SELECT ID, TALKID FROM T_CUS_SERV ; ";
                LOGGER.info("执行Sql操作V\n" + queryStr + "=>[]");
                conn.query(queryStr, res -> {
                    if (res.succeeded()) {
                        ResultSet rs = res.result();
                        List<JsonObject> resultJsonList = rs.getRows();
                        List<CustomerServ> csList = new ArrayList<>();
                        for(JsonObject row : resultJsonList) {
                            CustomerServ newCS = new CustomerServ();
                            newCS.setId(row.getLong("ID"));
                            newCS.setTalkId(row.getString("TALKID"));
                            csList.add(newCS);
                        }
                        if (csList.size() > 0) {
                            int randomNum = Constants.getRandomInt(csList.size());
                            randCSFut.complete(csList.get(randomNum));
                        } else {
                            randCSFut.fail("查找结果为空");
                        }
                        conn.close();
                    } else {
                        randCSFut.fail(res.cause());
                        conn.close();
                    }
                });
            } else {
                randCSFut.fail(as.cause());
            }
        });
        return randCSFut;
    }

    /**
     * 新增微信用户
     * @param wxUser
     * @return
     */
    public Future<WXUser> insertWxUser(WXUser wxUser) {
        Future insertWxUserFut = Future.future();
        StringBuilder insertWxUserSb = new StringBuilder();
        JsonArray params = new JsonArray()
                .add(wxUser.getWxOpenId() == null ? "" : wxUser.getWxOpenId())
                .add(wxUser.getHeadUrl() == null ? "" : wxUser.getHeadUrl())
                .add(wxUser.getWxNickName() == null ? "" : wxUser.getWxNickName())
                .add(wxUser.getSex() == null ? "" : wxUser.getSex())
                .add(wxUser.getUnionid() == null ? "" : wxUser.getUnionid())
                .add(wxUser.getCreateTime() == null ? "" : wxUser.getCreateTime());

        insertWxUserSb.append(" INSERT INTO T_WX_USER (WX_OPENID, HEAD_URL, WX_NICKNAME, SEX, UNIONID, CREATETIME) VALUES (?,?,?,?,?,?); ");
        insert(insertWxUserSb, params)
                .compose(id -> {
                    wxUser.setId(id);
                    insertWxUserFut.complete(wxUser);
                }, insertWxUserFut);
        return insertWxUserFut;
    }

    /**
     * 根据openId查找微信用户
     * @param openId
     * @return
     */
    public Future<WXUser> getWxUserByOpenId(String openId) {
        Future<WXUser> userInfoFut = Future.future();
        StringBuilder getWxUserInfoSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getWxUserInfoSb.append(" SELECT ID, WX_OPENID, HEAD_URL, WX_NICKNAME, SEX, UNIONID, CREATETIME FROM T_WX_USER WHERE WX_OPENID = ?; ");
        params.add(openId);

        query(getWxUserInfoSb, params)
                .compose(rows -> {
                    if (rows.size() == 0) {
                        userInfoFut.fail("不存在该用户[" + openId + "]");
                    } else {
                        JsonObject row = rows.get(0);
                        WXUser newUser = new WXUser();
                        newUser.setId(row.getLong("ID"))
                                .setWxOpenId(row.getString("WX_OPENID"))
                                .setHeadUrl(row.getString("HEAD_URL"))
                                .setWxNickName(row.getString("WX_NICKNAME"))
                                .setSex(row.getString("SEX"))
                                .setUnionid(row.getString("UNIONID"))
                                .setCreateTime(row.getString("CREATETIME"));
                        userInfoFut.complete(newUser);
                    }
                }, userInfoFut);
        return userInfoFut;
    }

    /**
     * 根据unionId查找微信用户
     * @param unionId
     * @return
     */
    public Future<WXUser> getWxUserByUnionId(String unionId) {
        Future<WXUser> userInfoFut = Future.future();
        StringBuilder getWxUserInfoSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getWxUserInfoSb.append(" SELECT ID, WX_OPENID, HEAD_URL, WX_NICKNAME, SEX, UNIONID, CREATETIME FROM T_WX_USER WHERE UNIONID = ?; ");
        params.add(unionId);

        query(getWxUserInfoSb, params)
                .compose(rows -> {
                    if (rows.size() == 0) {
                        userInfoFut.fail("不存在该用户[" + unionId + "]");
                    } else {
                        JsonObject row = rows.get(0);
                        WXUser newUser = new WXUser();
                        newUser.setId(row.getLong("ID"))
                                .setWxOpenId(row.getString("WX_OPENID"))
                                .setHeadUrl(row.getString("HEAD_URL"))
                                .setWxNickName(row.getString("WX_NICKNAME"))
                                .setSex(row.getString("SEX"))
                                .setUnionid(row.getString("UNIONID"))
                                .setCreateTime(row.getString("CREATETIME"));
                        userInfoFut.complete(newUser);
                    }
                }, userInfoFut);
        return userInfoFut;
    }

    /**
     *  获取指定时间内的聊天记录
     * @param openid
     * @param now           当前日期
     * @param before        30天前的日期
     * @return
     */
    public Future<List<MsgRecord>> getMsgRecordsOrderByDate(String openid, String now, String before) {
        Future<List<MsgRecord>> msgFut = Future.future();
        StringBuilder getMsgRecordsSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getMsgRecordsSb.append(" SELECT T.OPENID, T.CONTENT, T.CS_NICKNAME, T.CSID, T.SOURCE_TYPE, T.MSG_TYPE, DATE_FORMAT( T.SENDTIME, '%Y-%m-%d %H:%i:%S' ) AS SENDTIME FROM T_MSG_RECORD T ")
                .append(" WHERE T.SENDTIME <= ? ");
        params.add(now);
        if (StringUtils.isNotEmpty(before)) {
            getMsgRecordsSb.append(" AND T.SENDTIME > ? ");
            params.add(before);
        }
        if (StringUtils.isNotEmpty(openid)) {
            getMsgRecordsSb.append(" AND T.OPENID = ? ");
            params.add(openid);
        }
        getMsgRecordsSb.append(" ORDER BY T.SENDTIME DESC ; ");

        query(getMsgRecordsSb, params)
                .compose(rows -> {
                    List<MsgRecord> msgRecordsList = new ArrayList<>();
                    for(JsonObject row : rows) {
                        MsgRecord msg = new MsgRecord();
                        msg.setOpenId(row.getString("OPENID"))
                                .setContent(row.getString("CONTENT"))
                                .setSourceType(row.getString("SOURCE_TYPE"))
                                .setMsgType(row.getString("MSG_TYPE"))
                                .setCsNickName(row.getString("CS_NICKNAME"))
                                .setCsId(row.getString("CSID"))
                                .setSendtime(row.getLong("SENDTIME"));
                        msgRecordsList.add(msg);
                    }
                    msgFut.complete(msgRecordsList);
                }, msgFut);
        return msgFut;
    }

    /**
     * 以分页形式获取聊天记录
     * @param openId
     * @param startIndex
     * @param size
     * @return
     */
    public Future<List<MsgRecord>> getMsgRecordsOrderByDateInPage(String openId, Integer startIndex, Integer size) {
        Future<List<MsgRecord>> msgByPageFut = Future.future();
        StringBuilder getMsgRecordsByPageSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getMsgRecordsByPageSb.append(" SELECT T.OPENID, T.CONTENT, T.CS_NICKNAME, T.CSID, T.SOURCE_TYPE, T.MSG_TYPE, T.SENDTIME FROM T_MSG_RECORD T ")
                .append(" WHERE T.OPENID = ? ");
        params.add(openId);
        getMsgRecordsByPageSb.append(" ORDER BY T.SENDTIME DESC ")
                .append(" LIMIT ").append(startIndex).append(", ").append(size)
                .append("; ");

        query(getMsgRecordsByPageSb, params)
                .compose(rows -> {
                    List<MsgRecord> msgRecordsList = new ArrayList<>();
                    for(JsonObject row : rows) {
                        MsgRecord msg = new MsgRecord();
                        msg.setOpenId(row.getString("OPENID"))
                                .setContent(row.getString("CONTENT"))
                                .setSourceType(row.getString("SOURCE_TYPE"))
                                .setMsgType(row.getString("MSG_TYPE"))
                                .setCsNickName(row.getString("CS_NICKNAME"))
                                .setCsId(row.getString("CSID"))
                                .setSendtime(row.getLong("SENDTIME"));
                        msgRecordsList.add(msg);
                    }
                    msgByPageFut.complete(msgRecordsList);
                }, msgByPageFut);
        return msgByPageFut;
    }

    /**
     * 获取所有最近记录数
     * @param openId
     * @return
     */
    public Future<Integer> countAllMsgRecordByOpenid(String openId) {
        Future<Integer> msgCountFut = Future.future();
        JsonArray params = new JsonArray();
        StringBuilder countAllMsgRecSb = new StringBuilder("SELECT COUNT(0) AS COUNT FROM T_MSG_RECORD T ");
        countAllMsgRecSb.append(" WHERE T.OPENID = ? ; ");
        params.add(openId);

        query(countAllMsgRecSb, params)
                .compose(rows -> {
                    Integer countNum = 0;
                    if (rows.size() > 0) {
                        countNum = rows.get(0).getInteger("COUNT");
                        msgCountFut.complete(countNum);
                    } else {
                        msgCountFut.fail("不存在该用户[" + openId + "]");
                    }
                }, msgCountFut);
        return msgCountFut;
    }

    /**
     * 删除指定时间日期内的消息记录
     * @param now
     * @param before
     * @return
     */
    public Future<Void> deleteHistoryMsgRecored(String now, String before) {
        Future<Void> delHisMsgRecFut = Future.future();
        StringBuilder delHistoryMsgSb = new StringBuilder();
        JsonArray params = new JsonArray();
        delHistoryMsgSb.append(" DELETE FROM T_MSG_RECORD WHERE SENDTIME <= ? ");
        params.add(now);
        if (StringUtils.isNotEmpty(before)) {
            delHistoryMsgSb.append("AND T.SENDTIME > ? ");
            params.add(before);
        }
        delHistoryMsgSb.append(" ; ");

        update(delHistoryMsgSb, params).setHandler(delHisMsgRecFut.completer());
        return delHisMsgRecFut;
    }

    /**
     *  获取用户足记
     * @param openid
     * @return
     */
    public Future<List<UserAction>> getUserActionsByOpenId(String openid) {
        Future<List<UserAction>> actionFut = Future.future();

        StringBuilder getUserActionSb = new StringBuilder("SELECT TUA.START_TIME, TUA.END_TIME, TUA.ACT_TYPE, TUA.DESCRIPTION FROM T_USER_ACTION TUA WHERE TUA.OPENID = ? ORDER BY TUA.START_TIME DESC ;");
        JsonArray params = new JsonArray()
                .add(openid);

        query(getUserActionSb, params)
                .compose(rows -> {
                    List<UserAction> userActions = new ArrayList<>();
                    for(JsonObject row : rows) {
                        UserAction action = new UserAction();
                        action.setOpenId(openid)
                                .setActStartTime(row.getLong("START_TIME"))
                                .setActEndTime(row.getLong("END_TIME"))
                                .setActType(row.getString("ACT_TYPE"))
                                .setDescription(row.getString("DESCRIPTION"));
                        userActions.add(action);
                    }
                    actionFut.complete(userActions);
                }, actionFut);
        return actionFut;
    }

    /**
     * 获取公众号下用户互动限制信息
     * @param developerId
     * @return
     */
    public Future<List<UserLimit>> getUserLimitByDeveloperId(String developerId, Integer pageIndex, Integer pageSize) {
        Future<List<UserLimit>> userLimitListInDeveloFuture = Future.future();

        StringBuilder getUserLimitsSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getUserLimitsSb.append(" SELECT TUL.LAST_SEND_TIME AS SENDTIME, TUL.OPENID AS OPENID FROM T_USER_LIMIT TUL ")
                .append(" WHERE TUL.DEVELOPERID = ? ORDER BY TUL.LAST_SEND_TIME DESC LIMIT ?,?;  ");
        params.add(developerId)
                .add(pageIndex)
                .add(pageSize);

        query(getUserLimitsSb, params)
                .compose(rows -> {
                    List<UserLimit> limits = new ArrayList<>();
                    for (JsonObject row : rows) {
                        UserLimit newLimit = new UserLimit();
                        newLimit.setLaseSendTimestamp(row.getLong("SENDTIME"))
                                .setDeveloperId(developerId)
                                .setOpenId(row.getString("OPENID"));
                        limits.add(newLimit);
                    }
                    userLimitListInDeveloFuture.complete(limits);
                }, userLimitListInDeveloFuture);
        return userLimitListInDeveloFuture;
    }

    /**
     * 获取在线粉丝展示信息
     * @param developerId
     * @param pageIndex
     * @param pageSize
     * @return
     */
    public Future<List<RecentContactFans>> getOnlineFansByDeveloperId (String csId, String developerId, Integer pageIndex, Integer pageSize) {
        Future<List<RecentContactFans>> getOnlineFansFuture = Future.future();

        StringBuilder getOnlineFansSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getOnlineFansSb.append(" SELECT ")
                .append("TUL.LAST_SEND_TIME AS LASTACTTIMESTAMP, ")
                .append("TUL.OPENID AS OPENID, ")
                .append("TUL.DEVELOPERID, ")
                .append("TM.MSG_TYPE, ")
                .append("TUL.LIMIT_TEXT, ")
                .append("TM.CONTENT, ")
                .append("TM.SENDTIME ")
                .append("FROM T_USER_LIMIT TUL  ");
        if (Objects.nonNull(csId)) {
            // 客服下在线粉丝数据
            getOnlineFansSb
                    .append("INNER JOIN T_APP_USER TAU  ON TAU.WX_OPENID = TUL.OPENID ")
                    .append("LEFT JOIN T_MSG_RECORD TM ON TM.ID = TUL.LATELY_MSGID ")
                    .append(" WHERE TAU.CUSTOMER_SERVER_ID = ? ");
            params.add(csId);
        } else if (Objects.nonNull(developerId)) {
            // 公众号下在线粉丝数据
            getOnlineFansSb.append("LEFT JOIN T_MSG_RECORD TM ON TM.ID = TUL.LATELY_MSGID ")
                    .append(" WHERE TUL.DEVELOPERID = ? ");
            params.add(developerId);
        }
        if (pageSize != -1) {
            getOnlineFansSb.append(" ORDER BY TUL.LAST_SEND_TIME DESC LIMIT ?,?;  ");
            params.add(pageIndex)
                    .add(pageSize);
        }

        query(getOnlineFansSb, params)
                .compose(rows -> {
                    List<RecentContactFans> recentContactFans = new ArrayList<>();
                    for (JsonObject row : rows) {
                        RecentContactFans fansMsg = new RecentContactFans();
                        fansMsg.setPubName(row.getString("NAME"))
                                .setLimitText(row.getInteger("LIMIT_TEXT"))
                                .setDeveloperId(row.getString("DEVELOPERID"))
                                .setLastActTimestamp(row.getLong("LASTACTTIMESTAMP"))
                                .setLastSendTime(row.getLong("SENDTIME")==null?0L:row.getLong("SENDTIME"))
                                .setOpenId(row.getString("OPENID"));
                        String content = row.getString("CONTENT");
                        if (StringUtils.isNotBlank(content)) {
                            JsonObject contentJson = new JsonObject(content);

                            if (MsgType.IMAGE.getType().equals(row.getString("MSG_TYPE"))) {
                                fansMsg.setContent("[图片]");
                            } else if (MsgType.TEXT.getType().equals(row.getString("MSG_TYPE"))) {
                                fansMsg.setContent(EmojiUtil.decodeEmoji(contentJson.getString("content")));
                            }
                        } else {
                            fansMsg.setContent("");
                        }
                        recentContactFans.add(fansMsg);
                    }
                    getOnlineFansFuture.complete(recentContactFans);
                }, getOnlineFansFuture);
        return getOnlineFansFuture;
    }

    /**
     * 全渠道修改
     *      获取渠道下客服下的粉丝信息
     * @param csId
     * @param channelType
     * @param pageIndex
     * @param pageSize
     * @return
     */
    public Future<List<RecentContactFans>> getChannelOnlineFansByCS (String csId, Integer channelType, String name, Integer pageIndex, Integer pageSize) {
        Future<List<RecentContactFans>> getOnlineFansFuture = Future.future();

        StringBuilder getOnlineFansSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getOnlineFansSb.append(" SELECT ")
                .append("TUL.LAST_SEND_TIME AS LASTACTTIMESTAMP, ")
                .append("TUL.OPENID AS OPENID, ")
                .append("TUL.DEVELOPERID, ")
                .append("TM.MSG_TYPE, ")
                .append("TM.CONTENT, ")
                .append("TM.SENDTIME, ")
                .append("TUL.LIMIT_TEXT ")
                .append("FROM T_USER_LIMIT TUL  ");
        // innor join table
        if (Objects.nonNull(csId)) {
            // 客服下在线粉丝数据
            getOnlineFansSb
                    .append("INNER JOIN T_APP_USER TAU  ON TAU.WX_OPENID = TUL.OPENID ")
                    .append("LEFT JOIN T_MSG_RECORD TM ON TM.ID = TUL.LATELY_MSGID ");

        }
        if (Objects.nonNull(name)) {
            getOnlineFansSb.append(" INNER JOIN t_wx_user tu ON tu.wx_openid = TUL.openid ");
        }
        // where
        getOnlineFansSb.append(" WHERE 1 = 1 ");
        // where append
        if (Objects.nonNull(csId)) {
            getOnlineFansSb.append(" AND TAU.CUSTOMER_SERVER_ID = ? ");
            params.add(csId);
        }
        if (Objects.nonNull(channelType)) {
            getOnlineFansSb.append(" AND TAU.CHANNEL_TYPE = ? ");
            params.add(channelType);
        }
        if (Objects.nonNull(name)) {
            getOnlineFansSb.append(" AND tu.wx_nickname LIKE ? ");
            params.add("%" + name + "%");
        }
        // order by
        getOnlineFansSb.append(" ORDER BY TUL.LAST_SEND_TIME DESC ");
        // limit
        getOnlineFansSb.append(" LIMIT ?,?; ");
        params.add(pageIndex)
                .add(pageSize);

        query(getOnlineFansSb, params)
                .compose(rows -> {
                    List<RecentContactFans> recentContactFans = new ArrayList<>();
                    for (JsonObject row : rows) {
                        RecentContactFans fansMsg = new RecentContactFans();
                        fansMsg.setPubName(row.getString("NAME"))
                                .setLimitText(row.getInteger("LIMIT_TEXT"))
                                .setDeveloperId(row.getString("DEVELOPERID"))
                                .setLastActTimestamp(row.getLong("LASTACTTIMESTAMP"))
                                .setLastSendTime(row.getLong("SENDTIME")==null?0L:row.getLong("SENDTIME"))
                                .setOpenId(row.getString("OPENID"));
                        String content = row.getString("CONTENT");
                        if (StringUtils.isNotBlank(content)) {
                            JsonObject contentJson = new JsonObject(content);

                            if (MsgType.IMAGE.getType().equals(row.getString("MSG_TYPE"))) {
                                fansMsg.setContent("[图片]");
                            } else if (MsgType.TEXT.getType().equals(row.getString("MSG_TYPE"))) {
                                fansMsg.setContent(EmojiUtil.decodeEmoji(contentJson.getString("content")));
                            }
                        } else {
                            fansMsg.setContent("");
                        }
                        recentContactFans.add(fansMsg);
                    }
                    getOnlineFansFuture.complete(recentContactFans);
                }, getOnlineFansFuture);
        return getOnlineFansFuture;
    }

    public Future<List<RecentContactFans>> getChannelOnlineFansByCSChannelList (JsonArray csChannelList, String name, Integer pageIndex, Integer pageSize) {
        StringBuilder getOnlineFansSb = new StringBuilder();
        JsonArray params = new JsonArray();
        getOnlineFansSb.append(" SELECT ")
                .append("TUL.LAST_SEND_TIME AS LASTACTTIMESTAMP, ")
                .append("TUL.OPENID AS OPENID, ")
                .append("TUL.DEVELOPERID, ")
                .append("TM.MSG_TYPE, ")
                .append("TM.CONTENT, ")
                .append("TUL.LIMIT_TEXT, ")
                .append("TM.SENDTIME ")
                .append("FROM T_USER_LIMIT TUL  ")
                .append("INNER JOIN T_APP_USER TAU  ON TAU.WX_OPENID = TUL.OPENID ")
                .append("LEFT JOIN T_MSG_RECORD TM ON TM.ID = TUL.LATELY_MSGID ");
        if (Objects.nonNull(name)) {
            getOnlineFansSb.append(" INNER JOIN T_WX_USER TU ON TU.WX_OPENID = TUL.OPENID ");
        }
        // where
        getOnlineFansSb.append(" WHERE 1 = 1 ");
        // where append
        if (csChannelList != null && csChannelList.size() > 0) {
            getOnlineFansSb.append(" AND (");
            for (int index = 0; index < csChannelList.size(); index++) {
                if (0 != index) {
                    getOnlineFansSb.append(" OR ");
                }
                getOnlineFansSb.append(" (TAU.CUSTOMER_SERVER_ID = ? ")
                        .append(" AND TAU.CHANNEL_TYPE = ? ) ");
                params.add(csChannelList.getJsonObject(index).getInteger("csId"))
                        .add(csChannelList.getJsonObject(index).getInteger("channelType"));
            }
            getOnlineFansSb.append(" ) ");
        }
        if (Objects.nonNull(name)) {
            getOnlineFansSb.append(" AND TU.WX_NICKNAME LIKE ? ");
            params.add("%" + name + "%");
        }
        // order by
        getOnlineFansSb.append(" ORDER BY TUL.LAST_SEND_TIME DESC ");
        // limit
        getOnlineFansSb.append(" LIMIT ?,?; ");
        params.add(pageIndex)
                .add(pageSize);

        return Future.future(csChannelOnlineFansFut -> {
            query(getOnlineFansSb, params)
                    .compose(rows -> {
                        List<RecentContactFans> recentContactFans = new ArrayList<>();
                        for (JsonObject row : rows) {
                            RecentContactFans fansMsg = new RecentContactFans();
                            fansMsg.setPubName(row.getString("NAME"))
                                    .setLimitText(row.getInteger("LIMIT_TEXT"))
                                    .setDeveloperId(row.getString("DEVELOPERID"))
                                    .setLastActTimestamp(row.getLong("LASTACTTIMESTAMP"))
                                    .setLastSendTime(row.getLong("SENDTIME")==null?0L:row.getLong("SENDTIME"))
                                    .setOpenId(row.getString("OPENID"));
                            String content = row.getString("CONTENT");
                            if (StringUtils.isNotBlank(content)) {
                                JsonObject contentJson = new JsonObject(content);

                                if (MsgType.IMAGE.getType().equals(row.getString("MSG_TYPE"))) {
                                    fansMsg.setContent("[图片]");
                                } else if (MsgType.TEXT.getType().equals(row.getString("MSG_TYPE"))) {
                                    fansMsg.setContent(EmojiUtil.decodeEmoji(contentJson.getString("content")));
                                }
                            } else {
                                fansMsg.setContent("");
                            }
                            recentContactFans.add(fansMsg);
                        }
                        csChannelOnlineFansFut.complete(recentContactFans);
                    }, csChannelOnlineFansFut);
        });
    }

    /**
     * 获取客服下在线粉丝
     * @param csId
     * @return
     */
    public Future<List<UserLimit>> getUserLimitByCsId(String csId, Integer pageIndex, Integer pageSize) {
        Future<List<UserLimit>> userLimitListInCsFuture = Future.future();
        StringBuilder getUserLimitsSb = new StringBuilder();
        JsonArray params = new JsonArray();

        getUserLimitsSb
                .append(" SELECT TUL.OPENID, TUL.DEVELOPERID, TUL.LAST_SEND_TIME AS SENDTIME, TUL.LIMIT_TEXT AS TEXTLIMIT FROM T_APP_USER TAU INNER JOIN T_USER_LIMIT TUL ON TAU.WX_OPENID = TUL.OPENID WHERE TAU.CUSTOMER_SERVER_ID = ? ")
                .append(" ORDER BY TUL.LAST_SEND_TIME DESC ");
        params.add(csId);
        if (pageIndex == 0 && pageSize == -1) {
            // 获取全部
        } else {
            // 获取分页的
            getUserLimitsSb.append(" LIMIT ?,? ");
            params
                    .add(pageIndex)
                    .add(pageSize);
        }
        getUserLimitsSb.append(" ; ");

        query(getUserLimitsSb, params)
                .compose(rows -> {
                    List<UserLimit> limits = new ArrayList<>();
                    for (JsonObject row : rows) {
                        UserLimit newLimit = new UserLimit();
                        newLimit.setLaseSendTimestamp(row.getLong("SENDTIME"))
                                .setDeveloperId(row.getString("DEVELOPERID"))
                                .setTextLimit(row.getInteger("TEXTLIMIT"))
                                .setOpenId(row.getString("OPENID"));
                        limits.add(newLimit);
                    }
                    userLimitListInCsFuture.complete(limits);
                }, userLimitListInCsFuture);
        return userLimitListInCsFuture;
    }

    /**
     * 根据互动限制信息查询最后一条聊天记录
     * @param limits
     * @return
     */
    public Future<List<RecentContactFans>> getEveryUserLimitRecentSendInList(List<UserLimit> limits) {
        Future<List<RecentContactFans>> recentContactFansListFuture = Future.future();
        getEveryUserLimitRecentSend(limits).setHandler(recentContactFansListFuture);
        return recentContactFansListFuture;
    }

    /**
     * 获取每个用户的最近一条聊天记录
     *  TODO 该方法严重影响效率，待update
     * @param limits
     * @return
     */
    public Future<List<RecentContactFans>> getEveryUserLimitRecentSend(List<UserLimit> limits) {
        Future<List<RecentContactFans>> recentContactFansListFuture = Future.future();
        List<RecentContactFans> recentContactFans = new ArrayList<>();
        List<Future> futList = new ArrayList<>();

        // TODO 有待优化：batch操作
        for (UserLimit limit : limits) {
            Future<Void> lastSendFut = Future.future();
            futList.add(lastSendFut);
            StringBuilder queryLastSend = new StringBuilder();
            queryLastSend.append(" SELECT TM.MSG_TYPE, TM.CONTENT, TM.SENDTIME, TP.NAME FROM T_MSG_RECORD TM " )
                    .append(" INNER JOIN T_USER_LIMIT TUL ON TUL.OPENID = TM.OPENID INNER JOIN T_PUBLIC TP ON TP.DEVELOPERID = TUL.DEVELOPERID ")
                    .append(" WHERE TUL.OPENID = ? ORDER BY TM.SENDTIME DESC LIMIT 0, 1; ");
            JsonArray params = new JsonArray()
                    .add(limit.getOpenId());

            query(queryLastSend, params)
                    .compose(rows -> {
                        RecentContactFans fans = new RecentContactFans();
                        fans
                                .setOpenId(limit.getOpenId())
                                .setDeveloperId(limit.getDeveloperId())
                                .setLastActTimestamp(limit.getLaseSendTimestamp());
                        if (rows.size() > 0) {
                            fans
                                    .setPubName(rows.get(0).getString("NAME"))
                                    .setLastSendTime(rows.get(0).getLong("SENDTIME"));
                            if ("text".equals(rows.get(0).getString("MSG_TYPE"))) {
                                fans.setContent(rows.get(0).getString("CONTENT"));
                            } else if ("image".equals(rows.get(0).getString("MSG_TYPE"))) {
                                JsonObject contentJson = new JsonObject()
                                        .put("content", "[图片]");
                                fans.setContent(contentJson.encode());
                            }
                        }
                        recentContactFans.add(fans);
                        lastSendFut.complete();
                    }, lastSendFut);
        }
        CompositeFuture.all(futList)
                .compose((Void) -> {
                    recentContactFansListFuture.complete(recentContactFans);
                }, recentContactFansListFuture);
        return recentContactFansListFuture;
    }

    /**
     * 查询最近发送的一个消息
     * @param openid
     * @return
     */
    public Future<MsgRecord> findLastSendMsgByOpenId(String openid) {
        Future<MsgRecord> lastSendFut = Future.future();

        StringBuilder queryLastSend = new StringBuilder();
        JsonArray params = new JsonArray();
        queryLastSend.append(" SELECT ")
                .append("TM.ID, ")
                .append("TM.CONTENT, ")
                .append("TM.CSID, ")
                .append("TM.CS_NICKNAME, ")
                .append("TM.MSG_TYPE, ")
                .append("TM.OPENID, ")
                .append("TM.SENDTIME, ")
                .append("TM.SOURCE_TYPE, ")
                .append("TM.ISREAD ")
                .append("FROM ")
                .append("T_MSG_RECORD TM ")
                .append("WHERE ")
                .append("TM.OPENID = ? ")
                .append("ORDER BY ")
                .append("TM.SENDTIME DESC ")
                .append("LIMIT 0,1; ");
        params.add(openid);

        query(queryLastSend, params)
                .compose(rows -> {
                    MsgRecord msgRecord = new MsgRecord();
                    if (rows.size() > 0) {
                        JsonObject row = rows.get(0);
                        msgRecord.setMsgType(row.getString("MSG_TYPE"))
                                .setCsNickName(row.getString("CS_NICKNAME"))
                                .setCsId(row.getString("CSID"))
                                .setSendtime(row.getLong("SENDTIME"))
                                .setContent(row.getString("CONTENT"))
                                .setOpenId(row.getString("OPENID"))
                                .setSourceType(row.getString("SOURCE_TYPE"))
                                .setId(row.getLong("ID"))
                                .setIsRead(row.getString("ISREAD"));
                    } else {
                        // 无记录则返回空
                    }
                    lastSendFut.complete(msgRecord);
                }, lastSendFut);
        return lastSendFut;
    }

    /**
     * 更新用户互动限制
     * @param openId            // 微信用户唯一标识
     * @param updateMap         // 要更新的限制项键值对
     * @return
     */
    public Future<Void> updateUserLimit(String openId, EnumMap<ActiveLimit, Long> updateMap) {
        Future<Void> updateLimitFuture = Future.future();
        StringBuilder updateUserLimitSb = new StringBuilder();
        JsonArray params = new JsonArray();
        updateUserLimitSb.append(" UPDATE T_USER_LIMIT TUL SET ");

        StringBuilder updateColumn = new StringBuilder();
        for (ActiveLimit activeLimitKey : updateMap.keySet()) {
            switch (activeLimitKey) {
                case ACTIVE_LIMIT:
                    updateColumn.append(",TUL.LAST_SEND_TIME = ? ");
                    break;
                case  TEXT_LIMIT :
                    updateColumn.append(",TUL.LIMIT_TEXT = ? ");
                    break;
                case  PIC_LIMIT :
                    updateColumn.append(",TUL.LIMIT_PIC = ? ");
                    break;
                case  PIC_TEXT_LIMIT :
                    updateColumn.append(",TUL.LIMIT_PIC_TEXT = ? ");
                    break;
                case LATELY_MSG_ID:
                    updateColumn.append(",TUL.LATELY_MSGID = ? ");
                    break;
                default:
            }
            params.add(updateMap.get(activeLimitKey));
        }
        updateUserLimitSb.append(updateColumn.toString().substring(1))
                .append(" WHERE TUL.OPENID = ? ; ");
        params.add(openId);

        update(updateUserLimitSb, params).setHandler(updateLimitFuture.completer());
        return updateLimitFuture;
    }

    /**
     * 根据openid获取用户互动限制
     * @param openId
     * @return
     */
    public Future<UserLimit> getUserLimitByOpenId(String openId) {
        StringBuilder sqlSb = new StringBuilder();
        JsonArray params = new JsonArray();
        sqlSb.append("select * from t_user_limit ul where ul.openid = ?; ");
        params.add(openId);
        return Future.future(limitFut -> {
            query(sqlSb, params)
                    .compose(rows -> {
                        UserLimit ul = new UserLimit();
                        if (rows.size() > 0) {
                            JsonObject row = rows.get(0);
                            ul.setLatelyMsgId(row.getLong("lately_msgid"))
                                    .setDeveloperId(row.getString("developerid"))
                                    .setOpenId(row.getString("openid"))
                                    .setTextLimit(row.getInteger("limit_text"))
                                    .setLaseSendTimestamp(row.getLong("last_send_time"))
                                    .setId(row.getLong("id"));
                        } else {
                            // 无记录则返回空
                            ul = null;
                        }
                        limitFut.complete(ul);
                    }, limitFut);
        });
    }

    /**
     * 删除用户互动限制记录
     * @param openId
     * @return
     */
    public Future<Void> deleteUserLimit(String openId) {
        Future<Void> delLimitFuture = Future.future();
        StringBuilder delLimitSb = new StringBuilder();
        JsonArray params = new JsonArray();
        delLimitSb.append(" DELETE FROM T_USER_LIMIT WHERE OPENID = ?; ");
        params.add(openId);

        update(delLimitSb, params).setHandler(delLimitFuture.completer());
        return delLimitFuture;
    }

    /**
     * 更新客服服务数据信息
     * @param openId
     * @param isCount
     * @return
     */
    public Future<Void> updateCusServDataStateOrCountByOpenId(String openId, boolean isCount) {
        Future<Void> updateCsServiceDataFut = Future.future();
        StringBuilder updateCsServiceDataSb = new StringBuilder();
        JsonArray params = new JsonArray();
        updateCsServiceDataSb.append(" UPDATE T_CUS_SERV_DATA TCD SET ");
        if (isCount) {
            updateCsServiceDataSb.append(" TCD.ACTIVE_COUNT = (TCD.ACTIVE_COUNT + 1) ");
        } else {
            updateCsServiceDataSb.append(" TCD.CURRENT_USE = '1' ");
        }
        updateCsServiceDataSb.append(" WHERE TCD.OPENID = ?  ")
                .append(" AND TCD.CURRENT_USE = '0' ");
        params.add(openId);

        update(updateCsServiceDataSb, params).setHandler(updateCsServiceDataFut.completer());
        return updateCsServiceDataFut;
    }

    /**
     * 获取客服服务数据统计
     * @param developerId
     * @param csIdList
     * @return
     */
    public Future<List<CusServiceData>> countCusServDataByDeveloperIdOrCsIdList(String developerId, JsonArray csIdList, Integer pageIndex, Integer pageSize) {
        Future<List<CusServiceData>> countCsServiceFut = Future.future();
        List<CusServiceData> cusServiceDataList = new ArrayList<>();

        StringBuilder countCsServiceSb = new StringBuilder();
        JsonArray params = new JsonArray();
        countCsServiceSb.append(" SELECT SUM(TCD.ACTIVE_COUNT) AS COUNT, TCD.CSID AS CSID FROM T_CUS_SERV_DATA TCD ");
        if (StringUtils.isNotEmpty(developerId)) {
            countCsServiceSb.append(" WHERE TCD.DEVELOPERID = ?  ");
            params.add(developerId);
        }
        if (CodeUtil.collectionNotEmpty(csIdList)) {
            countCsServiceSb.append(" WHERE TCD.CSID IN ( ");
            for (Object csid : csIdList) {
                countCsServiceSb.append("?").append(",");
                params.add(csid);
            }
            countCsServiceSb.deleteCharAt(countCsServiceSb.length() - 1);
            countCsServiceSb.append(")");
        }
        countCsServiceSb.append(" GROUP BY TCD.CSID ")
                .append(" ORDER BY COUNT DESC LIMIT ?,? ; ");
        params.add(pageIndex)
                .add(pageSize);

        query(countCsServiceSb, params)
                .compose(rows -> {
                    for (JsonObject row : rows) {
                        CusServiceData cusServiceData = new CusServiceData();
                        cusServiceData.setCsId(row.getString("CSID"))
                                .setActiveCount(Long.valueOf(row.getString("COUNT")));
                        cusServiceDataList.add(cusServiceData);
                    }
                    countCsServiceFut.complete(cusServiceDataList);
                }, countCsServiceFut);
        return countCsServiceFut;
    }

    /**
     * 获取最近联系的用户
     * @param developerId
     * @param startTime
     * @param endTime
     * @return
     */
    public Future<List<CusServiceData>> getRecentlyActiveUser(String developerId, Long startTime, Long endTime) {
        Future<List<CusServiceData>> recentActiveFut = Future.future();
        StringBuilder querySb = new StringBuilder();
        JsonArray params = new JsonArray();
        querySb.append("");
        return recentActiveFut;
    }

    /**
     * 获取指定列表中粉丝的最近聊天记录
     * @param openIdList
     * @return
     */
    public Future<List<MsgRecord>> getLatelyMsgRecordByOpenId(JsonArray openIdList) {
        Future<List<MsgRecord>> msgListFut = Future.future();
        StringBuilder querySql = new StringBuilder();
        querySql.append(" SELECT TM.MSG_TYPE, TM.CONTENT, TM.SENDTIME, TUL.OPENID FROM T_USER_LIMIT TUL INNER JOIN T_MSG_RECORD TM ON TM.ID = TUL.LATELY_MSGID ")
                .append(" WHERE TUL.OPENID IN ( ");

        for (int i = 0; i < openIdList.size(); i++) {
            querySql.append("?").append(",");
        }
        querySql.deleteCharAt(querySql.length() - 1);
        querySql.append(" ); ");

        query(querySql, openIdList)
                .compose(msgList -> {
                    List<MsgRecord> msgRecordList = new ArrayList<>();
                    for (JsonObject row : msgList) {
                        MsgRecord newRecord = new MsgRecord();
                        newRecord.setOpenId(row.getString("OPENID"))
                                .setMsgType(row.getString("MSG_TYPE"))
                                .setContent(row.getString("CONTENT"))
                                .setSendtime(row.getLong("SENDTIME"));
                        msgRecordList.add(newRecord);
                    }
                    msgListFut.complete(msgRecordList);
                }, msgListFut);
        return msgListFut;
    }

    /**
     * 获取姓名模糊查询的粉丝
     * @param developerId
     * @param name
     * @param page
     * @return
     */
    public Future<List<RecentContactFans>> getOnlineFansByName(String developerId, String name, PageInfo page) {
        Future<List<RecentContactFans>> findFansByNameFuture = Future.future();
        StringBuilder querySb = new StringBuilder();
        JsonArray params = new JsonArray();
        querySb.append(" SELECT ")
                .append("TM.MSG_TYPE, ")
                .append("TM.CONTENT, ")
                .append("TM.SENDTIME, ")
                .append("TUL.LAST_SEND_TIME, ")
                .append("TUL.OPENID, ")
                .append("TUL.LIMIT_TEXT ")
                .append("FROM ")
                .append(" T_USER_LIMIT TUL ")
                .append("INNER JOIN T_WX_USER TU ON TU.WX_OPENID = TUL.OPENID ")
                .append("INNER JOIN T_MSG_RECORD TM ON TM.OPENID = TUL.OPENID ")
                .append("AND TM.ID = TUL.LATELY_MSGID ")
                .append("WHERE ")
                .append("TU.WX_NICKNAME LIKE ? AND TUL.DEVELOPERID = ? ");
        params.add("%" + name + "%")
                .add(developerId);
        if (!(page.getPageNum() == 0 && page.getPageSize() == -1)) {
            querySb.append(" ORDER BY TUL.LAST_SEND_TIME DESC LIMIT ?,?; ");
            params.add(page.getCurrentIndex())
                    .add(page.getPageSize());
        }

        query(querySb, params)
                .compose(msgList -> {
                    List<RecentContactFans> recentContactFans = new ArrayList<>();
                    for (JsonObject row : msgList) {
                        RecentContactFans newRecentContactFans = new RecentContactFans();
                        if (MsgType.IMAGE.getType().equals(row.getString("MSG_TYPE"))) {
                            JsonObject contentJson = new JsonObject();
                            contentJson.put("content", "[图片]");
                            newRecentContactFans.setContent(contentJson.encode());
                        } else if (MsgType.TEXT.getType().equals(row.getString("MSG_TYPE"))) {
                            newRecentContactFans.setContent(row.getString("CONTENT"));
                        }
                        newRecentContactFans.setOpenId(row.getString("OPENID"))
                                .setLastSendTime(row.getLong("SENDTIME"))
                                .setLimitText(row.getInteger("LIMIT_TEXT"))
                                .setLastActTimestamp(row.getLong("LAST_SEND_TIME"));
                        recentContactFans.add(newRecentContactFans);
                    }
                    findFansByNameFuture.complete(recentContactFans);
                }, findFansByNameFuture);
        return findFansByNameFuture;
    }

    /**
     * 获取最近(30天)联系粉丝
     * @param csId           客服下最近联系粉丝
     * @param developerId    公众号下最近联系
     * @param name           昵称模糊查询条件(为null则不需要昵称模糊查询)
     * @param page
     * @return
     */
    public Future<List<RecentContactFans>> getRecentContactFans(String csId, String developerId, String name, PageInfo page) {
        Future<List<RecentContactFans>> findFansByDeveloperFuture = Future.future();
        StringBuilder querySb = new StringBuilder();
        JsonArray params = new JsonArray();
        querySb.append(" SELECT TM.MSG_TYPE, TM.CONTENT, TM.SENDTIME, TUL.LAST_SEND_TIME, TUL.OPENID FROM T_USER_LIMIT TUL ")
                .append(" INNER JOIN T_MSG_RECORD TM ON TM.OPENID = TUL.OPENID AND TM.ID = TUL.LATELY_MSGID ");
        // 插入连表
        if (StringUtils.isNotBlank(name)) {
            querySb.append(" INNER JOIN T_WX_USER TU ON TU.WX_OPENID = TUL.OPENID ");
        }
        // 插入where 连接
        querySb.append(" WHERE 1 = 1 ");
        if (StringUtils.isNotBlank(developerId)) {
            querySb.append(" AND TUL.DEVELOPERID = ? AND TUL.LAST_SEND_TIME > ? ");
            params.add(developerId)
                    .add(PeriodicTime.get30DaysBeforeTimestamp());
        }
        if (StringUtils.isNotBlank(name)) {
            querySb.append(" AND TU.WX_NICKNAME LIKE ? ");
            params.add("%" + name + "%");
        }
        if (StringUtils.isNotBlank(csId)) {
            querySb.append(" AND TM.CSID = ? ");
            params.add(csId);
        }
        // 0，-1查询全部
        if (!(page.getPageNum() == 0 && page.getPageSize() == -1)) {
            // 排序和分页信息
            querySb.append(" ORDER BY TUL.LAST_SEND_TIME DESC LIMIT ?,?;  ");
            params.add(page.getCurrentIndex())
                    .add(page.getPageSize());
        }

        query(querySb, params)
                .compose(msgList -> {
                    List<RecentContactFans> recentContactFans = new ArrayList<>();
                    for (JsonObject row : msgList) {
                        RecentContactFans newRecentContactFans = new RecentContactFans();
                        if (MsgType.IMAGE.getType().equals(row.getString("MSG_TYPE"))) {
                            JsonObject contentJson = new JsonObject();
                            contentJson.put("content", "[图片]");
                            newRecentContactFans.setContent(contentJson.encode());
                        } else if (MsgType.TEXT.getType().equals(row.getString("MSG_TYPE"))) {
                            newRecentContactFans.setContent(row.getString("CONTENT"));
                        }
                        newRecentContactFans.setOpenId(row.getString("OPENID"))
                                .setLastSendTime(row.getLong("SENDTIME"))
                                .setLastActTimestamp(row.getLong("LAST_SEND_TIME"));
                        recentContactFans.add(newRecentContactFans);
                    }
                    findFansByDeveloperFuture.complete(recentContactFans);
                }, findFansByDeveloperFuture);
        return findFansByDeveloperFuture;
    }

    /**
     * 清数据调用的接口
     *      TODO 该方法会清楚所有数据，慎用
     * @return
     */
    public Future<Void> deleteAllContentInTableForClearData() {
        Future<Void> delAllFut = Future.future();
        List<String> tableList = new ArrayList<>();
        tableList.add("T_APP_USER");
        tableList.add("T_CUS_SERV");
        tableList.add("T_CUS_SERV_DATA");
        tableList.add("T_MSG_RECORD");
//        tableList.add("T_PUBLIC");
        tableList.add("T_TOKEN");
        tableList.add("T_USER_ACTION");
        tableList.add("T_USER_LIMIT");
        tableList.add("T_WX_USER");
        List<Future> futList = new ArrayList<>();
        for (String tab : tableList) {
            Future<Void> delTabFut = Future.future();
            futList.add(delTabFut);
            StringBuilder delTableSb = new StringBuilder();
            delTableSb.append(" DELETE FROM ")
                    .append(tab)
                    .append(";");
            update(delTableSb, new JsonArray())
                    .setHandler(delTabFut.completer());
        }
        CompositeFuture.all(futList)
                .compose(ar -> {
                    delAllFut.complete();
                }, delAllFut);
        return delAllFut;
    }

    /**
     * 微博拉黑/取消拉黑时更新拉黑状态
     * @param uidList
     * @param pullBlack
     * @return
     */
    public Future<Void> setWBPullBlackStatus(List<String> uidList, boolean pullBlack) {
        StringBuilder updateSql = new StringBuilder();
        JsonArray params = new JsonArray();
        updateSql.append(" UPDATE t_app_user ")
                .append(" SET pull_black = ? ")
                .append(" WHERE ")
                .append(" wx_openid IN ( ");
        params.add(pullBlack?1:0);
        for (String uid : uidList) {
            updateSql.append("?").append(",");
            params.add(uid);
        }
        updateSql.deleteCharAt(updateSql.length() - 1);
        updateSql.append("); ");
        return update(updateSql, params);
    }

    /**
     * 更新web粉丝在线状态
     * @param openId
     * @param isOnline
     * @return
     */
    public Future<Void> updateWebFansOnlineStatus(String openId, boolean isOnline) {
        StringBuilder updateSql = new StringBuilder();
        JsonArray params = new JsonArray();
        updateSql.append(" UPDATE T_USER_LIMIT ")
                .append(" SET LIMIT_TEXT = ? ")
                .append(" WHERE ")
                .append(" OPENID = ? ");
        params.add(openId).add(isOnline?100:0);

        return update(updateSql, params);
    }

    /**
     * 更新web粉丝talkId
     * @param phone
     * @param oldTalkId
     * @param newTalkId
     * @return
     */
    public Future<Void> updateWebFansTalkByPhone(String phone, String oldTalkId, String newTalkId) {
        StringBuilder updateSql = new StringBuilder();
        JsonArray params = new JsonArray();
        updateSql.append(" UPDATE t_wx_user ")
                .append(" SET unionid = ?")
                .append(" WHERE ")
                .append(" wx_openid = ? ")
                .append(" AND unionid = ? ;");
        params.add(newTalkId).add(phone).add(oldTalkId);

        return update(updateSql, params);
    }

    //-----------------------------------------------------------------------------------------
    //  通用基本查询 [注意：vert.x所有batch操作暂不支持]
    //-----------------------------------------------------------------------------------------
    /**
     * query
     * @param querySql
     * @param params
     * @return
     */
    public Future<List<JsonObject>> query(StringBuilder querySql, JsonArray params) {
        Future<List<JsonObject>> queryResFut = Future.future();

        LOGGER.info("执行query操作->\n{} => param=[{}]", querySql, params);
        getConnection()
                .compose(conn -> {
                    if (CodeUtil.collectionNotEmpty(params)) {
                        conn.queryWithParams(querySql.toString(),
                                params,
                                queryAr -> {
                                    conn.close();
                                    if (queryAr.failed()) {
                                        queryResFut.fail(queryAr.cause());
                                    } else {
                                        List<JsonObject> rows = queryAr.result().getRows();
                                        queryResFut.complete(rows);
                                    }
                                });
                    } else {
                        conn.query(querySql.toString(),
                                queryAr -> {
                                    conn.close();
                                    if (queryAr.failed()) {
                                        queryResFut.fail(queryAr.cause());
                                    } else {
                                        List<JsonObject> rows = queryAr.result().getRows();
                                        queryResFut.complete(rows);
                                    }
                                });
                    }
                }, queryResFut);
        return queryResFut;
    }

    /**
     * update
     * @param updateSql
     * @param params
     * @return
     */
    public Future<Void> update(StringBuilder updateSql, JsonArray params) {
        Future<Void> updateResFut = Future.future();

        LOGGER.info("执行update操作->\n{} => param=[{}]", updateSql, params);
        getConnection()
                .compose(conn -> {
                    conn.updateWithParams(updateSql.toString(),
                            params,
                            updateAr -> {
                                conn.close();
                                if (updateAr.failed()) {
                                    updateResFut.fail(updateAr.cause());
                                } else {
                                    updateResFut.complete();
                                }
                            });
                }, updateResFut);
        return updateResFut;
    }

    /**
     * 通用新增
     * {"keys":[1413],"updated":1}
     * @param updateSql
     * @param params
     * @return
     */
    public Future<Long> insert(StringBuilder updateSql, JsonArray params) {
        Future<Long> insertResFut = Future.future();

        LOGGER.info("执行insert操作->\n{} => param=[{}]", updateSql, params);
        getConnection()
                .compose(conn -> {
                    conn.updateWithParams(updateSql.toString(),
                            params,
                            insertAr -> {
                                conn.close();
                                if (insertAr.failed()) {
                                    insertResFut.fail(insertAr.cause());
                                } else {
                                    JsonObject returnJson = insertAr.result().toJson();
                                    JsonArray key = returnJson.getJsonArray("keys");
                                    insertResFut.complete(key.getLong(0));
                                }
                            });
                }, insertResFut);
        return insertResFut;
    }
    //-----------------------------------------------------------------------------------------
    //  通用实体查询
    //-----------------------------------------------------------------------------------------
    /**
     *  新增表实体
     * @param t
     * @param <T>
     */
    public <T> Future<T> save(T t) {
        Future<T> saveFut = Future.future();
        try {
            // 分析注解信息获取数据库中表结构
            Class<?> clazz = t.getClass();
            List<String> columnList = new ArrayList<>();
            JsonArray jsonArray = new JsonArray();
            APPTable tableNameAnno = clazz.getAnnotation(APPTable.class);
            String tableName = tableNameAnno.table();

            String idFieldName = "";
            Field[] fields = clazz.getDeclaredFields();
            for (Field field : fields) {
                APPTable columnAnno;
                if ((columnAnno = field.getAnnotation(APPTable.class)) != null) {
                    if (columnAnno.isId()) {
                        idFieldName = field.getName();
                        continue;
                    }
                    String fieldName = field.getName();
                    fieldName = fieldName.substring(0,1).toUpperCase() + fieldName.substring(1);
                    // 调用get方法
                    Method method = t.getClass().getMethod("get" + fieldName);
                    Object methodValue = method.invoke(t);
                    columnList.add(columnAnno.column());
                    if (methodValue == null) {
                        if (field.getType() == String.class) {
                            jsonArray.add("");
                        } else if (field.getType() == Long.class) {
                            jsonArray.add(0L);
                        } else if (field.getType() == Integer.class) {
                            jsonArray.add(0);
                        }
                    } else {
                        jsonArray.add(methodValue);
                    }
                }
            }

            // 执行插入语句
            final String idFinal = idFieldName;

            StringBuilder insertSb = new StringBuilder();
            StringBuilder columnSb = new StringBuilder();
            StringBuilder paramSb = new StringBuilder();

            for (String column : columnList) {
                columnSb.append(column).append(",");
                paramSb.append("?").append(",");
            }
            columnSb.deleteCharAt(columnSb.length() - 1);
            paramSb.deleteCharAt(paramSb.length() - 1);

            insertSb.append("insert into ").append(tableName)
                    .append(" ( ")
                    .append(columnSb.toString())
                    .append(" ) ")
                    .append("values ( ")
                    .append(paramSb.toString())
                    .append(" ); ");

            insert(insertSb, jsonArray)
                    .compose(id -> {
                        // 设置id值
                        String idStr = idFinal.substring(0, 1).toUpperCase() + idFinal.substring(1);
                        try{
                            Method idMethod = t.getClass().getDeclaredMethod("set" + idStr, Long.class);
                            idMethod.invoke(t, id);
                            saveFut.complete(t);
                        } catch (Exception invokExcp) {
                            invokExcp.printStackTrace();
                            saveFut.fail(invokExcp.getCause());
                        }
                    }, saveFut);
        } catch (Exception e) {
            e.printStackTrace();
            saveFut.fail(e);
        }
        return saveFut;
    }

    /**
     * 更新表实体
     * @param t
     * @param <T>
     */
    public <T> Future<Void> update(T t) {
        Future<Void> updateFut = Future.future();
        try {
            // 分析注解信息获取数据库中表结构
            Class<?> clazz = t.getClass();
            List<String> columnList = new ArrayList<>();
            JsonArray jsonArray = new JsonArray();
            APPTable tableNameAnno = clazz.getAnnotation(APPTable.class);
            String tableName = tableNameAnno.table();

            String idFieldName = "";
            Object id = null;
            Field[] fields = clazz.getDeclaredFields();
            for (Field field : fields) {
                APPTable columnAnno;
                if ((columnAnno = field.getAnnotation(APPTable.class)) != null) {
                    String fieldName = field.getName();
                    fieldName = fieldName.substring(0, 1).toUpperCase() + fieldName.substring(1);
                    try {
                        Method method = t.getClass().getMethod("get" + fieldName);
                        Object methodValue = method.invoke(t);

                        if (columnAnno.isId()) {
                            idFieldName = columnAnno.column();
                            id = methodValue;
                            continue;
                        }
                        columnList.add(columnAnno.column());
                        if (methodValue == null) {
                            if (field.getType() == String.class) {
                                jsonArray.add("");
                            } else if (field.getType() == Long.class) {
                                jsonArray.add(0L);
                            } else if (field.getType() == Integer.class) {
                                jsonArray.add(0);
                            }
                        } else {
                            jsonArray.add(methodValue);
                        }
                        // jsonArray.add(methodValue);
                    } catch (Exception nsme) {
                        nsme.printStackTrace();
                        updateFut.fail(nsme);
                        return null;
                    }
                }
            }
            // id属性放在参数最后
            jsonArray.add(id);
            // 拼接更新的表项
            StringBuilder columnSb = new StringBuilder();
            for(String column : columnList) {
                columnSb.append(column).append("=?,");
            }
            columnSb.deleteCharAt(columnSb.length() - 1);
            StringBuilder updateSb = new StringBuilder();
            updateSb.append("update ").append(tableName)
                    .append(" set ")
                    .append(columnSb.toString())
                    .append(" where ")
                    .append(idFieldName)
                    .append(" = ?; ");

            update(updateSb, jsonArray)
                    .setHandler(updateFut.completer());
        } catch (Exception e) {
            e.printStackTrace();
            updateFut.fail(e.getCause());
        }
        return updateFut;

    }

    /**
     * 根据实体主键ID获取实体
     * @param id    [注意：Long类型]
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> Future<T> get(Long id, Class<T> clazz) {

        Future<T> getFut = Future.future();
        try {
            // 分析注解信息获取数据库中表结构
            Map<String, String> columnMap = new HashMap<>();

            APPTable tableNameAnno = clazz.getAnnotation(APPTable.class);
            String tableName = tableNameAnno.table();

            String idFieldName = "";
            Field[] fields = clazz.getDeclaredFields();
            for (Field field : fields) {
                APPTable columnAnno;
                if ((columnAnno = field.getAnnotation(APPTable.class)) != null) {
                    if (columnAnno.isId()) {
                        idFieldName = columnAnno.column();
                    }
                    // 映射 成员变量和表项
                    columnMap.put(columnAnno.column(), field.getName());
                }
            }

            StringBuilder querySb = new StringBuilder();
            JsonArray jsonArray = new JsonArray()
                    .add(id);
            querySb.append("select * from ").append(tableName)
                    .append(" where ")
                    .append(idFieldName)
                    .append(" = ?; ");

            query(querySb, jsonArray)
                    .compose(rows -> {
                        JsonObject resultJson = rows.get(0);
                        JsonObject columnToVarJson = new JsonObject();
                        for(String key : columnMap.keySet()) {
                            columnToVarJson.put(columnMap.get(key), resultJson.getValue(key));
                        }
                        T entity = Json.decodeValue(columnToVarJson.toString(), clazz);
                        getFut.complete(entity);
                    }, getFut);
        } catch (Exception e) {
            e.printStackTrace();
            getFut.fail(e);
        }
        return getFut;
    }

    /**
     * 对实体删除
     * @param id
     * @param clazz
     * @return
     */
    public Future<Void> delete(Long id, Class<?> clazz) {
        Future<Void> deleteFut = Future.future();
        try {
            // 分析注解信息获取数据库中表结构
            APPTable tableNameAnno = clazz.getAnnotation(APPTable.class);
            String tableName = tableNameAnno.table();

            Field[] fields = clazz.getDeclaredFields();
            String idName = "";
            for (Field field : fields) {
                APPTable columnAnno;
                if ((columnAnno = field.getAnnotation(APPTable.class)) != null) {
                    if (columnAnno.isId()) {
                        idName = columnAnno.column();
                        break;
                    }
                }
            }

            // 执行sql
            final String idNameFinal = idName;
            StringBuilder deleteSb = new StringBuilder();
            JsonArray jsonArray = new JsonArray()
                    .add(id);
            deleteSb.append("delete from ").append(tableName)
                    .append(" where ")
                    .append(idNameFinal)
                    .append(" = ?; ");
            update(deleteSb, jsonArray)
                    .setHandler(deleteFut.completer());
        } catch (Exception e) {
            e.printStackTrace();
            deleteFut.fail(e);
        }
        return deleteFut;
    }


    public Future<Void> testFut() {
        Future<Void> updateResFut = Future.future();

        LOGGER.info("执行testFut");

        String query = "select * from t_wx_user;";
        getConnection()
                .compose(conn -> {
                    Future someOperation1 = Future.future();
                    conn.queryWithParams(query, new JsonArray(), someOperation1);
                    someOperation1
                            .compose(ar -> {
                                ResultSet rs = (ResultSet)ar;
                                List<JsonObject> jsonList = rs.getRows();
                                Future someOperation2 = Future.future();
                                conn.queryWithParams(query, new JsonArray(), someOperation2);
                                return someOperation2;
                            })
                            .compose(ar2 -> {
                                ResultSet rs = (ResultSet)ar2;
                                List<JsonObject> jsonList = rs.getRows();
                                Future someOperation3 = Future.future();
                                conn.queryWithParams(query, new JsonArray(), someOperation3);
                                return someOperation3;
                            }).compose(ar3 -> {
                                ResultSet rs = (ResultSet)ar3;
                                List<JsonObject> jsonList = rs.getRows();
                                Future someOperationEnd = Future.future();
                                conn.queryWithParams(query, new JsonArray(), someOperationEnd);
                                return someOperationEnd;
                            });
        }, updateResFut);
        return updateResFut;
    }

    public static void main(String[] args) throws Exception {

        WXUser wxUser = new WXUser();
        wxUser
                .setSex("0")
                .setUnionid("123456789")
                .setWxNickName("andaren")
                .setHeadUrl("http://5f5s5f")
                .setWxOpenId("fds55df48d4");
        //save(wxUser);
    }

}
